package edu.hawaii.ics.pjdb.worker.command_runners;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;

import org.apache.thrift.TException;
import org.apache.thrift.server.TServer;
import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import edu.hawaii.ics.pjdb.result_transport.TupleListener;
import edu.hawaii.ics.pjdb.result_transport.TupleServiceHelper;
import edu.hawaii.ics.pjdb.types.DbSyncCommand;
import edu.hawaii.ics.pjdb.types.Tuple;
import edu.hawaii.ics.pjdb.types.WorkerCommand;

public class DbSyncRunner extends CommandRunner {
    private static final Logger logger = LoggerFactory
            .getLogger(DbSyncRunner.class);

    /**
     * Handles input tuples and executes the queries and discards the query
     * results (results should not be expected from DbSync commands).
     * 
     */
    protected static class SyncTupleListener extends TupleListener {
        private static final Logger logger = LoggerFactory
                .getLogger(SyncTupleListener.class);

        private final DbSyncCommand cmd;
        private final Connection localDb;
        private String sql;
        private boolean sqlSet;

        public SyncTupleListener(WorkerCommand workerCmd, Connection localDb)
                throws SQLException, TTransportException, InterruptedException {
            this.localDb = localDb;
            this.cmd = workerCmd.getCommand().getDbSyncCmd();
            this.sql = "INSERT INTO " + cmd.getTableName() + " VALUES(";
            this.sqlSet = false;
            logger.info("Stuff is all set up. Listening for tuples.");
        }

        @Override
        public void receiveTuples(List<Tuple> tuples) {
            logger.debug("#receiveTuples() called.");

            try {
                // If the full SQL string has not been set yet, generate it
                if (!sqlSet) {
                    // Generate SQL string to insert values
                    int numParameters = tuples.get(0).getEntriesSize();
                    for (int i = 0; i < numParameters; i++) {

                        // Last parameter, close VALUES statement
                        if (i == numParameters - 1) {
                            this.sql = this.sql + "?)";
                        }

                        // Otherwise "," to prepare for next parameter
                        else {
                            this.sql = this.sql + "?,";
                        }
                    }
                }

                PreparedStatement statement = localDb
                        .prepareStatement(this.sql);

                for (Tuple tuple : tuples) {
                    logger.info("Received tuple: {}", tuple.getEntries());

                    // Bind parameters
                    int parameterIndex = 1;
                    for (String fieldValue : tuple.getEntries()) {
                        statement.setObject(parameterIndex, fieldValue);
                        parameterIndex++;
                    }

                    // Execute
                    statement.execute();
                }
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }

        @Override
        public void finished() {
            logger.info("#finish() called.");
        }

    }

    public DbSyncRunner(Connection localDb) {
        super(localDb);
    }

    @Override
    public void execute(WorkerCommand cmd) throws TException, SQLException {
        logger.info(
                "Received DbSync command. Setting up input listener on port {}",
                cmd.getCommand().getDbSyncCmd().getInputAddress().getPort());

        try {
            SyncTupleListener syncWorker = new SyncTupleListener(cmd, localDb);
            TServer inputServer = TupleServiceHelper
                    .createNonThreadedServer(cmd.getCommand().getDbSyncCmd()
                            .getInputAddress().getPort(), syncWorker);
            inputServer.serve();
        } catch (InterruptedException e) {
            throw new TException(e);
        }
    }
}
